package omega

import (
	"context"
	"fmt"
	"hash/crc32"
	"time"

	"gitee.com/zawei/yggdrasil/pkg/lib/omega/proto"
	"gitee.com/zawei/yggdrasil/pkg/log"
	"github.com/pkg/errors"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

type transport struct {
	TransportConfig
	compensateCmdChan    []chan *proto.GrpcCompensateCommand
	tccCoordinateCmdChan []chan *proto.GrpcTccCoordinateCommand
	txCli                proto.TxEventServiceClient
	tccCli               proto.TccEventServiceClient
}

func newGRPCClient(config GrpcClientConfig) (*grpc.ClientConn, error) {
	var ctx = context.Background()
	var dialOptions = config.dialOptions

	// 默认配置使用block
	if config.Block {
		if config.DialTimeout > time.Duration(0) {
			var cancel context.CancelFunc
			ctx, cancel = context.WithTimeout(ctx, config.DialTimeout)
			defer cancel()
		}

		dialOptions = append(dialOptions, grpc.WithBlock())
	}

	if config.KeepAlive != nil {
		dialOptions = append(dialOptions, grpc.WithKeepaliveParams(*config.KeepAlive))
	}

	if config.Credentials.Enable {
		creds, err := credentials.NewClientTLSFromFile(
			config.Credentials.CertFile, config.Credentials.ServerName,
		)
		if err != nil {
			return nil, err
		}
		dialOptions = append(dialOptions, grpc.WithTransportCredentials(creds))
	} else {
		dialOptions = append(dialOptions, grpc.WithInsecure())
	}
	if config.BalancerName != "" {
		dialOptions = append(dialOptions,
			grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"loadBalancingPolicy":"%s"}`, config.BalancerName)),
		)
	}

	return grpc.DialContext(ctx, config.Target, dialOptions...)
}

func newTransport(c TransportConfig) (*transport, error) {
	if c.MaxPayloadSize == 0 {
		c.MaxPayloadSize = 10240
	}
	if c.TransportTimeout == 0 {
		c.TransportTimeout = 5 * time.Second
	}
	if c.TransportReconDelay == 0 {
		c.TransportReconDelay = 1 * time.Second
	}
	if c.CompCmdChanNum == 0 {
		c.CompCmdChanNum = 32
	}
	if c.TccCmdChanNum == 0 {
		c.TccCmdChanNum = 32
	}
	if c.InstanceId == "" {
		c.InstanceId = fmt.Sprintf("%s-%s", c.ServiceName, getUuid())
	}

	t := &transport{
		TransportConfig:      c,
		compensateCmdChan:    make([]chan *proto.GrpcCompensateCommand, c.CompCmdChanNum),
		tccCoordinateCmdChan: make([]chan *proto.GrpcTccCoordinateCommand, c.TccCmdChanNum),
	}
	for i := 0; i < int(c.CompCmdChanNum); i++ {
		t.compensateCmdChan[i] = make(chan *proto.GrpcCompensateCommand, 2)
	}
	for i := 0; i < int(c.TccCmdChanNum); i++ {
		t.tccCoordinateCmdChan[i] = make(chan *proto.GrpcTccCoordinateCommand, 2)
	}
	grpcClient, err := newGRPCClient(c.GrcpClient)
	if err != nil {
		return nil, err
	}
	t.txCli = proto.NewTxEventServiceClient(grpcClient)
	t.tccCli = proto.NewTccEventServiceClient(grpcClient)
	log.Infof("omega transport, service_name[%s], instance[%s]", t.ServiceName, t.InstanceId)
	return t, nil
}

func (t *transport) getCompensateCmdChan() []chan *proto.GrpcCompensateCommand {
	return t.compensateCmdChan
}

func (t *transport) getTccCoordinateCmdChan() []chan *proto.GrpcTccCoordinateCommand {
	return t.tccCoordinateCmdChan
}

func (t *transport) handleTx(ctx context.Context) error {
	cli := t.txOnConnected(ctx)
	go func() {
		for {
			cmd, err := cli.Recv()
			if err != nil {
				select {
				case <-ctx.Done():
					return
				default:
					log.Errorf("fault to receive saga cmd, error: %s", err.Error())
					if err := t.txOnDisconnected(); err != nil {
						log.Errorf("fault to disconnected tx, error: %s", err.Error())
					}
					cli = t.txOnConnected(ctx)
				}
				continue
			}
			idx := crc32.ChecksumIEEE([]byte(cmd.GlobalTxId)) % t.CompCmdChanNum
			t.compensateCmdChan[idx] <- cmd
		}
	}()

	return nil
}

func (t *transport) txOnConnected(ctx context.Context) proto.TxEventService_OnConnectedClient {
	for {
		cli, err := t.txCli.OnConnected(ctx, &proto.GrpcServiceConfig{
			ServiceName: t.ServiceName,
			InstanceId:  t.InstanceId,
		})
		if err != nil {
			log.Errorf("fault to connected tx event service, error: %s", err.Error())
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(t.TransportReconDelay):
				continue
			}
		}
		return cli
	}
}

func (t *transport) txOnDisconnected() error {
	ctx, cancel := context.WithTimeout(context.TODO(), t.TransportTimeout)
	defer cancel()
	_, err := t.txCli.OnDisconnected(ctx, &proto.GrpcServiceConfig{
		ServiceName: t.ServiceName,
		InstanceId:  t.InstanceId,
	})
	return err
}

func (t *transport) txOnTxEvent(ctx context.Context, e *proto.GrpcTxEvent) (bool, error) {
	omegaCtx := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	e.GlobalTxId = omegaCtx.GlobalTxId
	e.LocalTxId = omegaCtx.LocalTxId
	e.ParentTxId = omegaCtx.ParentTxId
	e.ServiceName = t.ServiceName
	e.InstanceId = t.InstanceId
	e.Timestamp = getTimestamp()
	ctx, cancel := context.WithTimeout(ctx, t.TransportTimeout)
	defer cancel()
	log.Debugf("tx event[%s]", log.Json(e))
	ack, err := t.txCli.OnTxEvent(ctx, e)
	if err != nil {
		log.Errorf("fault to send tx event[%s], error: %s", err.Error())
		return false, errors.WithStack(err)
	}
	if ack.Aborted {
		log.Errorf("send tx event[%s] aborted", log.Json(e))
	}
	return ack.Aborted, nil
}

func (t *transport) sendTxAbortedEvent(ctx context.Context, compensationMethod string, err error) (bool, error) {
	errStr := err.Error()
	if len(errStr) > int(t.MaxPayloadSize) {
		errStr = errStr[0:t.MaxPayloadSize]
	}
	e := &proto.GrpcTxEvent{
		Type:               TxAbortedEvent,
		CompensationMethod: compensationMethod,
		Payloads:           []byte(errStr),
	}
	return t.txOnTxEvent(ctx, e)
}

func (t *transport) sendSagaAbortedEvent(ctx context.Context, compensationMethod string, err error) (bool, error) {
	errStr := err.Error()
	if len(errStr) > int(t.MaxPayloadSize) {
		errStr = errStr[0:t.MaxPayloadSize]
	}
	e := &proto.GrpcTxEvent{
		Type:               SagaAbortedEvent,
		CompensationMethod: compensationMethod,
		Payloads:           []byte(errStr),
	}
	return t.txOnTxEvent(ctx, e)
}

//----------------------------------------------------------------

func (t *transport) handleTcc(ctx context.Context) error {
	cli := t.tccOnConnected(ctx)
	go func() {
		for {
			cmd, err := cli.Recv()
			if err != nil {
				select {
				case <-ctx.Done():
					return
				default:
					log.Errorf("fault receive tcc cmd, error: %s", err.Error())
					if err := t.tccOnDisconnected(); err != nil {
						log.Errorf("fault disconnected tcc, error: %s", err.Error())
					}
					cli = t.tccOnConnected(ctx)
				}
				continue
			}
			idx := crc32.ChecksumIEEE([]byte(cmd.GlobalTxId)) % t.TccCmdChanNum
			t.tccCoordinateCmdChan[idx] <- cmd
		}
	}()

	return nil
}

func (t *transport) tccOnConnected(ctx context.Context) proto.TccEventService_OnConnectedClient {
	for {
		cli, err := t.tccCli.OnConnected(ctx, &proto.GrpcServiceConfig{
			ServiceName: t.ServiceName,
			InstanceId:  t.InstanceId,
		})
		if err != nil {
			log.Errorf("fault to connected tcc event service, error: %s", err.Error())
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(t.TransportReconDelay):
				continue
			}
		}
		return cli
	}
}

func (t *transport) tccOnDisconnected() error {
	ctx, cancel := context.WithTimeout(context.TODO(), t.TransportTimeout)
	defer cancel()
	_, err := t.tccCli.OnDisconnected(ctx, &proto.GrpcServiceConfig{
		ServiceName: t.ServiceName,
		InstanceId:  t.InstanceId,
	})
	return err
}

func (t *transport) OnParticipationStarted(ctx context.Context, confirmMethod, cancelMethod string) (bool, error) {
	omegaCtx := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	e := &proto.GrpcParticipationStartedEvent{
		Timestamp:     getTimestamp(),
		GlobalTxId:    omegaCtx.GlobalTxId,
		LocalTxId:     omegaCtx.LocalTxId,
		ParentTxId:    omegaCtx.ParentTxId,
		ServiceName:   t.ServiceName,
		InstanceId:    t.InstanceId,
		ConfirmMethod: confirmMethod,
		CancelMethod:  cancelMethod,
	}
	ctx, cancel := context.WithTimeout(ctx, t.TransportTimeout)
	defer cancel()
	log.Debugf("participation started, event[%s]", log.Json(e))
	ack, err := t.tccCli.OnParticipationStarted(ctx, e)
	if err != nil {
		log.Errorf("fault to send participation started, event[%s], error: %s", log.Json(e), err.Error())
		return false, errors.WithStack(err)
	}
	if ack.Aborted {
		log.Errorf("send send participation started aborted, event[%s]", log.Json(e))
	}
	return ack.Aborted, nil
}

func (t *transport) OnParticipationEnded(ctx context.Context, confirmMethod, cancelMethod, status string) (bool, error) {
	omegaCtx := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	e := &proto.GrpcParticipationEndedEvent{
		Timestamp:     getTimestamp(),
		GlobalTxId:    omegaCtx.GlobalTxId,
		LocalTxId:     omegaCtx.LocalTxId,
		ParentTxId:    omegaCtx.ParentTxId,
		ServiceName:   t.ServiceName,
		InstanceId:    t.InstanceId,
		ConfirmMethod: confirmMethod,
		CancelMethod:  cancelMethod,
		Status:        status,
	}
	ctx, cancel := context.WithTimeout(ctx, t.TransportTimeout)
	defer cancel()
	log.Debugf("participation ended, event[%s]", log.Json(e))
	ack, err := t.tccCli.OnParticipationEnded(ctx, e)
	if err != nil {
		log.Errorf("fault to send participation ended: event[%s], error: %s", log.Json(e), err.Error())
		return false, errors.WithStack(err)
	}
	if ack.Aborted {
		log.Errorf("send send participation ended aborted, event[%s]", log.Json(e))
	}
	return ack.Aborted, nil
}

func (t *transport) OnTccTransactionStarted(ctx context.Context) (bool, error) {
	omegaCtx := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	e := &proto.GrpcTccTransactionStartedEvent{
		Timestamp:   getTimestamp(),
		GlobalTxId:  omegaCtx.GlobalTxId,
		LocalTxId:   omegaCtx.LocalTxId,
		ParentTxId:  "",
		ServiceName: t.ServiceName,
		InstanceId:  t.InstanceId,
	}
	ctx, cancel := context.WithTimeout(ctx, t.TransportTimeout)
	defer cancel()
	log.Debugf("tcc transaction started, event[%s]", log.Json(e))
	ack, err := t.tccCli.OnTccTransactionStarted(ctx, e)
	if err != nil {
		log.Errorf("fault to send tcc transaction started: event[%s], error: %s", log.Json(e), err.Error())
		return false, errors.WithStack(err)
	}
	if ack.Aborted {
		log.Errorf("send tcc transaction started event aborted, event[%s]", log.Json(e))
	}
	return ack.Aborted, nil
}

func (t *transport) OnTccTransactionEnded(ctx context.Context, status string) (bool, error) {
	omegaCtx := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	e := &proto.GrpcTccTransactionEndedEvent{
		Timestamp:   getTimestamp(),
		GlobalTxId:  omegaCtx.GlobalTxId,
		LocalTxId:   omegaCtx.LocalTxId,
		ParentTxId:  omegaCtx.ParentTxId,
		ServiceName: t.ServiceName,
		InstanceId:  t.InstanceId,
		Status:      status,
	}
	ctx, cancel := context.WithTimeout(ctx, t.TransportTimeout)
	defer cancel()
	log.Debugf("tcc transaction ended, event[%s]", log.Json(e))
	ack, err := t.tccCli.OnTccTransactionEnded(ctx, e)
	if err != nil {
		log.Errorf("fault to send tcc transaction ended: event[%s], error: %s", log.Json(e), err.Error())
		return false, errors.WithStack(err)
	}
	if ack.Aborted {
		log.Errorf("send tcc transaction ended event aborted, event[%s]", log.Json(e))
	}
	return ack.Aborted, nil
}

func (t *transport) OnTccCoordinated(ctx context.Context, methodName, status string) (bool, error) {
	omegaCtx := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	e := &proto.GrpcTccCoordinatedEvent{
		Timestamp:   getTimestamp(),
		GlobalTxId:  omegaCtx.GlobalTxId,
		LocalTxId:   omegaCtx.LocalTxId,
		ParentTxId:  omegaCtx.ParentTxId,
		ServiceName: t.ServiceName,
		InstanceId:  t.InstanceId,
		MethodName:  methodName,
		Status:      status,
	}
	ctx, cancel := context.WithTimeout(ctx, t.TransportTimeout)
	defer cancel()
	log.Debugf("tcc coordinated, event[%s]", log.Json(e))
	ack, err := t.tccCli.OnTccCoordinated(ctx, e)
	if err != nil {
		log.Errorf("fault to send tcc coordinated: event[%s], error: %s", log.Json(e), err.Error())
		return false, errors.WithStack(err)
	}
	if ack.Aborted {
		log.Errorf("send tcc coordinated event aborted, event[%s]", log.Json(e))
	}
	return ack.Aborted, nil
}
